package com.weng.bigdata.framework.jdbc.utils;

import cn.hutool.core.convert.Convert;
import cn.hutool.core.date.DateTime;
import cn.hutool.db.ds.simple.SimpleDataSource;
import com.weng.bigdata.core.database.enums.DatabaseTypeEnum;
import com.weng.bigdata.core.jdbc.config.CustomDataSourceConfig;
import com.weng.bigdata.core.jdbc.model.meta.*;
import com.weng.bigdata.core.jdbc.utils.DateJdbcUtils;
import com.weng.bigdata.core.jdbc.enums.JdbcToJavaMappingEnum;
import com.weng.bigdata.core.jdbc.enums.SqlTypeDefaultLengthEnum;
import com.weng.bigdata.core.jdbc.enums.SqlTypeEnum;
import com.weng.framework.common.utils.collection.CollectionUtils;
import com.weng.framework.core.constants.GlobalConstants;
import com.weng.framework.core.constants.ResultCodeEnum;
import com.weng.framework.core.exception.DataSourceException;
import com.weng.framework.core.exception.GlobalServiceException;
import lombok.extern.slf4j.Slf4j;
import net.sf.jsqlparser.JSQLParserException;
import net.sf.jsqlparser.parser.CCJSqlParserUtil;
import net.sf.jsqlparser.statement.select.PlainSelect;
import net.sf.jsqlparser.statement.select.Select;
import org.apache.commons.lang3.StringUtils;
import org.springframework.jdbc.core.JdbcTemplate;
import org.springframework.jdbc.core.PreparedStatementCreator;
import org.springframework.jdbc.core.RowCallbackHandler;
import org.springframework.jdbc.support.JdbcUtils;

import javax.sql.DataSource;
import java.math.BigDecimal;
import java.sql.*;
import java.util.*;
import java.util.regex.Matcher;
import java.util.regex.Pattern;


@Slf4j
public class JdbcUtil {

    private static  int resultLimit = 1000000;

    private  static boolean isQueryLogEnable = false;

    private static final String TABLE = "TABLE";

    private static final String VIEW = "VIEW";

    private static final String[] TABLE_TYPES = new String[]{TABLE, VIEW};

    private static final String TABLE_NAME = "TABLE_NAME";

    private static final String TABLE_TYPE = "TABLE_TYPE";

    private static final String QUERY_COUNT_SQL = "SELECT COUNT(*) FROM (%s) CT";

    private static final String QUERY_META_SQL = "SELECT * FROM (%s) MT WHERE 1=0";

    /**
     * 敏感sql操作
     */
    private static final String REG_SENSITIVE_SQL = "drop\\s|alter\\s|grant\\s|insert\\s|replace\\s|delete\\s|truncate\\s|update\\s|remove\\s";


    /**
     * 匹配多行sql注解正则
     */
    private static final String REG_SQL_ANNOTATE = "(?ms)('(?:''|[^'])*')|--.*?$|/\\*[^+]*?\\*/";



    public static  void execute(String sql, String jdbcUrl, String username, String password) throws GlobalServiceException {
        sql = filterAnnotate(sql);
        checkSensitiveSql(sql);
        if (isQueryLogEnable) {
            log.info("{}", sql);
        }
        try {
            jdbcTemplate(jdbcUrl,  username, password).execute(sql);
        } catch (Exception e) {
            throw new GlobalServiceException(ResultCodeEnum.SQL_EXCEPTION.getCode(), e.getMessage());
        }
    }

    public static PageWithQueryColumnsMeta syncQuery4Paginate(String sql, Integer pageNo, Integer pageSize, Integer totalCount, Integer limit, Set<String> excludeColumns
            , String jdbcUrl, String username, String password, DatabaseTypeEnum databaseTypeEnum) throws Exception {
        if (null == pageNo) {
            pageNo = -1;
        }
        if (null == pageSize) {
            pageSize = -1;
        }
        if (null == totalCount) {
            totalCount = 0;
        }

        if (null == limit) {
            limit = -1;
        }

        PageWithQueryColumnsMeta paginate = query4Paginate(sql, pageNo, pageSize, totalCount, limit, excludeColumns, jdbcUrl, username, password, databaseTypeEnum);
        return paginate;
    }

    public static  List<Map<String, Object>> query4List(String sql, int limit,String jdbcUrl, String username, String password) throws Exception {
        sql = filterAnnotate(sql);
        checkSensitiveSql(sql);
        String md5 = MD5JdbcUtil.getMD5(sql, true, 16);
        if (isQueryLogEnable) {
            log.info("{}  >> \n{}", md5, sql);
        }
        JdbcTemplate jdbcTemplate = jdbcTemplate(jdbcUrl, username, password);
        jdbcTemplate.setMaxRows(limit > resultLimit ? resultLimit : limit);

        long befor = System.currentTimeMillis();

        List<Map<String, Object>> list = jdbcTemplate.queryForList(sql);

        if (isQueryLogEnable) {
            log.info("{} query for >> {} ms", md5, System.currentTimeMillis() - befor);
        }

        return list;
    }

    /**
     *  要获取remark信息，还需要加上额外属性支持，如jdbc:mysql://x.x.x.x:3306/djypt_test?remarks=true&useInformationSchema=true
     * @param jdbcUrl
     * @return
     */
    private static String setJdbcOfRemarkProperties(String jdbcUrl){
        StringBuilder jdbcUrlSb = new StringBuilder(jdbcUrl.trim());
        String remarkProperty = "remarks=true&useInformationSchema=true";
        if (jdbcUrlSb.indexOf("?") < 0) {
            jdbcUrlSb.append("?");
        }
        if (jdbcUrlSb.indexOf("&") > 0) {
            jdbcUrlSb.append("&");
        }
        jdbcUrlSb.append(remarkProperty);
        return jdbcUrlSb.toString();
    }

    /**
     * 获取数据库连接
     * @param jdbcUrl, driver
     * @param loginTimeout: Sets the maximum time in seconds that a driver will wait]
     * @return
     */
    public static Connection getConn(String jdbcUrl,String userName, String password, String driver, Integer loginTimeout) throws ClassNotFoundException, SQLException {
        Class.forName(driver);
        // 超时时间设置
        DriverManager.setLoginTimeout(loginTimeout);
        return DriverManager.getConnection(jdbcUrl,userName, password);
    }

    public  static PageWithQueryColumnsMeta query4Paginate(String sql, int pageNo, int pageSize, int totalCount, int limit, Set<String> excludeColumns
            , String jdbcUrl, String username, String password, DatabaseTypeEnum dataTypeEnum) throws Exception {
        PageWithQueryColumnsMeta paginateWithQueryColumns = new PageWithQueryColumnsMeta();
        sql = filterAnnotate(sql);
        checkSensitiveSql(sql);

        String md5 = MD5JdbcUtil.getMD5(sql + pageNo + pageSize + limit, true, 16);

        long before = System.currentTimeMillis();

        JdbcTemplate jdbcTemplate = jdbcTemplate(jdbcUrl, username, password);
        jdbcTemplate.setMaxRows(resultLimit);

        if (pageNo < 1 && pageSize < 1) {

            if (limit > 0) {
                resultLimit = limit > resultLimit ? resultLimit : limit;
            }
            if (isQueryLogEnable) {
                log.info("{}  >> \n{}", md5, sql);
            }
            jdbcTemplate.setMaxRows(resultLimit);
            getResultForPaginate(sql, paginateWithQueryColumns, jdbcTemplate, excludeColumns, -1);
            paginateWithQueryColumns.setCurPage(1);
            int size = paginateWithQueryColumns.getPageData().size();
            paginateWithQueryColumns.setPageSize(size);
            paginateWithQueryColumns.setTotalRecords(size);
        } else {
            paginateWithQueryColumns.setCurPage(pageNo);
            paginateWithQueryColumns.setPageSize(pageSize);

            final int startRow = (pageNo - 1) * pageSize;

            if (pageNo == 1 || totalCount == 0) {
                Object o = jdbcTemplate.queryForObject(getCountSql(sql), Object.class);
                totalCount = Integer.parseInt(String.valueOf(o));
            }
            if (limit > 0) {
                limit = limit > resultLimit ? resultLimit : limit;
                totalCount = limit < totalCount ? limit : totalCount;
            }

            paginateWithQueryColumns.setTotalRecords(totalCount);
            int maxRows = limit > 0 && limit < pageSize * pageNo ? limit : pageSize * pageNo;

            switch (dataTypeEnum) {
                case MYSQL:
                    sql = sql + " LIMIT " + startRow + ", " + pageSize;
                    md5 = MD5JdbcUtil.getMD5(sql, true, 16);
                    if (isQueryLogEnable) {
                        log.info("{}  >> \n{}", md5, sql);
                    }
                    getResultForPaginate(sql, paginateWithQueryColumns, jdbcTemplate, excludeColumns, -1);
                    break;
                default:
                    if (isQueryLogEnable) {
                        log.info("{}  >> \n{}", md5, sql);
                    }
                    jdbcTemplate.setMaxRows(maxRows);
                    getResultForPaginate(sql, paginateWithQueryColumns, jdbcTemplate, excludeColumns, startRow);
                    break;
            }
        }

        if (isQueryLogEnable) {
            log.info("{} query for >> {} ms", md5, System.currentTimeMillis() - before);
        }

        return paginateWithQueryColumns;
    }

    public static void getMetaDataBySql(String sql, String jdbcUrl, String username, String password) throws SQLException {

        List<Map<String, Object>> list = new ArrayList<>();
        try(Connection connection = getDataSource(jdbcUrl, username, password).getConnection();
            Statement stmt =connection.createStatement();
            ResultSet rs = stmt.executeQuery(sql)) {
                ResultSetMetaData metaData = rs.getMetaData();
                for(int i=1; i<= metaData.getColumnCount(); i++){
                    Map<String, Object> map = new HashMap<>();
                    map.put("getColumnName", metaData.getColumnName(i));
                    map.put("getColumnLabel", metaData.getColumnLabel(i));
                    map.put("getColumnType", metaData.getColumnType(i));
                    map.put("getColumnTypeName", metaData.getColumnTypeName(i));
                    map.put("getPrecision", metaData.getPrecision(i));
                    map.put("getTableName",metaData.getTableName(i));
                    map.put("getCatalogName", metaData.getCatalogName(i));
                    map.put("getColumnDisplaySize", metaData.getColumnDisplaySize(i));
                    map.put("getScale", metaData.getScale(i));
                    map.put("getSchemaName", metaData.getSchemaName(i));
                    list.add(map);
                }
        }
        System.out.println(list);
    }

    private  static void getResultForPaginate(String sql, PageWithQueryColumnsMeta paginateWithQueryColumns, JdbcTemplate jdbcTemplate, Set<String> excludeColumns, int startRow) {
        jdbcTemplate.query(sql, rs -> {
            if (null != rs) {
                ResultSetMetaData metaData = rs.getMetaData();

                List<SimpleColumnMeta> queryColumns = new ArrayList<>();
                for (int i = 1; i <= metaData.getColumnCount(); i++) {
                    String key = metaData.getColumnLabel(i);
                    if (!CollectionUtils.isEmpty(excludeColumns) && excludeColumns.contains(key)) {
                        continue;
                    }
                    queryColumns.add(new SimpleColumnMeta(key, metaData.getColumnTypeName(i)));
                }
                paginateWithQueryColumns.setColumns(queryColumns);

                List<Map<String, Object>> resultList = new ArrayList<>();

                try {
                    if (startRow > 0) {
                        rs.absolute(startRow);
                    }
                    while (rs.next()) {
                        resultList.add(getResultObjectMap(excludeColumns, rs, metaData));
                    }
                } catch (Throwable e) {
                    int currentRow = 0;
                    while (rs.next()) {
                        if (currentRow >= startRow) {
                            resultList.add(getResultObjectMap(excludeColumns, rs, metaData));
                        }
                        currentRow++;
                    }
                }

                paginateWithQueryColumns.setPageData(resultList);
            }
            return paginateWithQueryColumns;
        });
    }

    private  static  Map<String, Object> getResultObjectMap(Set<String> excludeColumns, ResultSet rs, ResultSetMetaData metaData) throws SQLException {
        Map<String, Object> map = new LinkedHashMap<>();
        for (int i = 1; i <= metaData.getColumnCount(); i++) {
            String key = metaData.getColumnLabel(i);
            if (!CollectionUtils.isEmpty(excludeColumns) && excludeColumns.contains(key)) {
                continue;
            }
            map.put(key, rs.getObject(key));
        }
        return map;
    }

    public static String getCountSql(String sql) {
        try {
            sql = "select * from (" + sql + ") temp";
            Select select = (Select) CCJSqlParserUtil.parse(sql);
            PlainSelect plainSelect = (PlainSelect) select.getSelectBody();
            plainSelect.setOrderByElements(null);
            return String.format(QUERY_COUNT_SQL, select.toString());
        } catch (JSQLParserException e) {
        }
        return String.format(QUERY_COUNT_SQL, sql);
    }


    /**
     * 获取当前连接数据库
     *
     * @return
     * @throws DataSourceException
     */
    public static  List<String> getDatabases(String username, String jdbcUrl, String password
            , DatabaseTypeEnum dataTypeEnum) throws DataSourceException {
        List<String> dbList = new ArrayList<>();
        Connection connection = null;
        try {
            connection = getConnection(jdbcUrl, username, password);
            if (null != connection) {
                switch (dataTypeEnum) {
                    case ORACLE:
                        dbList.add(username);
                        break;
                    default:
                        String catalog = connection.getCatalog();
                        if (!StringUtils.isEmpty(catalog)) {
                            dbList.add(catalog);
                        } else {
                            DatabaseMetaData metaData = connection.getMetaData();
                            ResultSet rs = metaData.getCatalogs();
                            while (rs.next()) {
                                dbList.add(rs.getString(1));
                            }
                        }
                        break;
                }
            }
        } catch (Exception e) {
            throw new DataSourceException(e.getMessage() + ", jdbcUrl=" + jdbcUrl);
        } finally {
            releaseConnection(connection);
        }
        return dbList;
    }

    /**
     * 获取当前数据源所有表结构
     *
     * @return
     * @throws DataSourceException
     */
    public static List<SimpleTableMeta> getSimpleTableMetaList(String jdbcUrl, String username, String password) throws DataSourceException {
        List<SimpleTableMeta> tableList = null;
        Connection connection = null;
        try {
            connection = getConnection(jdbcUrl, username, password);
            if (null != connection) {
                DatabaseMetaData metaData = connection.getMetaData();
                tableList = getSimpleTableMetaListByDatabaseMetaData(jdbcUrl, username, metaData);
            }
        } catch (Exception e) {
            throw new DataSourceException(e.getMessage() + ", jdbcUrl=" + jdbcUrl);
        } finally {
            releaseConnection(connection);
        }
        return tableList;
    }

    /**
     * 获取当前数据源表结构
     *
     * @return
     * @throws DataSourceException
     */
    public static List<SimpleTableMeta> getSimpleTableMetaListByDatabaseMetaData(String jdbcUrl, String username, DatabaseMetaData metaData)
            throws DataSourceException, SQLException {
        String dbName = getSchemaByJdbcUrl(jdbcUrl);
        List<SimpleTableMeta> tableList = null;
        String schema = null;
        try {
            schema = metaData.getConnection().getSchema();
        } catch (Throwable t) {
            log.error(t.getMessage(), t);
        }
        ResultSet tables = metaData.getTables(dbName, getDBSchemaPattern(schema,username,jdbcUrl), "%", TABLE_TYPES);
        if (null != tables) {
            tableList = new ArrayList<>();
            while (tables.next()) {
                String name = tables.getString(TABLE_NAME);
                if (!StringUtils.isEmpty(name)) {
                    String type = TABLE;
                    try {
                        type = tables.getString(TABLE_TYPE);
                    } catch (Exception e) {
                    }
                    String remark = "";
                    try {
                        remark = tables.getString("REMARKS");
                    } catch (Exception e) {
                    }
                    tableList.add(new SimpleTableMeta(name, type, remark));
                }
            }
            tables.close();
        }
        return tableList;
    }

    private  static String getDBSchemaPattern(String schema, String username, String jdbcUrl) {
        String schemaPattern = null;
        DatabaseTypeEnum dataTypeEnum = DatabaseTypeEnum.urlOf(jdbcUrl);
        if (null != dataTypeEnum) {
            switch (dataTypeEnum) {
                case ORACLE:
                    schemaPattern = username;
                    if (null != schemaPattern) {
                        schemaPattern = schemaPattern.toUpperCase();
                    }
                    break;
                case SQLSERVER:
                    schemaPattern = "dbo";

                case PRESTO:
                    if (!StringUtils.isEmpty(schema)) {
                        schemaPattern = schema;
                    }
                    break;
            }
        }
        return schemaPattern;

    }

    /**
     * 获取某个库所有表、包括表信息、列元数据的详细信息
     *
     * @return
     * @throws DataSourceException
     */
    public static List<TableMeta> getAllTableDetails(String jdbcUrl,   String username, String password) throws DataSourceException {
        List<TableMeta> tableMetaList = new ArrayList<>();
        Connection connection = null;
        try {
            connection = getConnection(jdbcUrl, username, password);
            if (null != connection) {
                DatabaseMetaData metaData = connection.getMetaData();
                List<SimpleTableMeta> tableList = getSimpleTableMetaListByDatabaseMetaData(jdbcUrl, username, metaData);
                for(SimpleTableMeta simpleTableMeta : tableList){
                    TableMeta tableMeta = Convert.convert(TableMeta.class, simpleTableMeta);
                    tableMetaList.add(setColumnsByMetaData(tableMeta, jdbcUrl, simpleTableMeta.getName(), metaData));
                }
                return tableMetaList;
            }
        } catch (SQLException e) {
            throw new DataSourceException(e.getMessage() + ", jdbcUrl=" + jdbcUrl);
        } finally {
            releaseConnection(connection);
        }
        return tableMetaList;
    }


    /**
     * 获取指定表列信息
     *
     * @param tableName
     * @return
     * @throws DataSourceException
     */
    public static TableMeta getTableInfo(String jdbcUrl, String tableName, String username, String password) throws DataSourceException {
        Connection connection = null;
        try {
            connection = getConnection(jdbcUrl, username, password);
            if (null != connection) {
                DatabaseMetaData metaData = connection.getMetaData();
                TableMeta tableMeta = new TableMeta();
                tableMeta.setName(tableName);
                return setColumnsByMetaData(tableMeta, jdbcUrl, tableName, metaData);
            }
        } catch (SQLException e) {
            throw new DataSourceException(e.getMessage() + ", jdbcUrl=" + jdbcUrl);
        } finally {
            releaseConnection(connection);
        }
        return null;
    }

    /**
     * 获取指定表列信息
     *
     * @param tableName
     * @return
     * @throws DataSourceException
     */
    private static TableMeta setColumnsByMetaData(TableMeta tableMeta,String jdbcUrl,  String tableName,DatabaseMetaData metaData) throws DataSourceException {
        String dbName = getSchemaByJdbcUrl(jdbcUrl);
        List<String> primaryKeys = getPrimaryKeys(dbName, tableName, metaData);
        List<ColumnMeta> columns = getColumns(tableName, metaData, jdbcUrl);
        tableMeta.setColumns(columns);
        tableMeta.setPrimaryKeys(primaryKeys);
        return tableMeta;
    }


    /**
     * 判断表是否存在
     *
     * @param tableName
     * @return
     * @throws DataSourceException
     */
    public boolean tableIsExist(String tableName, String jdbcUrl, String username, String password) throws DataSourceException {
        boolean result = false;
        Connection connection = null;
        try {
            connection = getConnection(jdbcUrl, username, password);
            if (null != connection) {
                ResultSet tables = connection.getMetaData().getTables(null, null, tableName, null);
                if (null != tables && tables.next()) {
                    result = true;
                    tables.close();
                } else {
                    result = false;
                }
            }
        } catch (Exception e) {
            throw new DataSourceException("Get connection meta data error, jdbcUrl=" + jdbcUrl);
        } finally {
            releaseConnection(connection);
        }

        return result;
    }

    /**
     * 根据sql查询列
     *
     * @param sql
     * @return
     * @throws DataSourceException
     */
    public List<SimpleColumnMeta> getColumnsBySql(String sql, String jdbcUrl, String username, String password) throws DataSourceException {
        long l = System.currentTimeMillis();
        checkSensitiveSql(sql);
        List<SimpleColumnMeta> columnList = new ArrayList<>();
        Connection connection = null;
        try {
            connection = getConnection(jdbcUrl, username, password);
            if (null != connection) {
                try (Statement statement = connection.createStatement();) {
                    statement.setMaxRows(1);
                    try (ResultSet resultSet = statement.executeQuery(sql);) {
                        ResultSetMetaData rsmd = resultSet.getMetaData();
                        int columnCount = rsmd.getColumnCount();
                        for (int i = 1; i <= columnCount; i++) {
                            SimpleColumnMeta queryColumn = new SimpleColumnMeta(
                                    rsmd.getColumnLabel(i),
                                    SqlTypeEnum.getType(rsmd.getColumnType(i)));
                            columnList.add(queryColumn);
                        }
                    }
                }
            }


        } catch (Exception e) {
            throw new DataSourceException(e.getMessage());
        } finally {
            releaseConnection(connection);
        }
        long l1 = System.currentTimeMillis();
        log.info("get columns for >>> {} ms", l1 - l);
        return columnList;
    }


    /**
     * 获取数据表主键
     *
     * @param tableName
     * @param metaData
     * @return
     * @throws GlobalServiceException
     */
    private static  List<String> getPrimaryKeys(String dbName, String tableName, DatabaseMetaData metaData) throws GlobalServiceException {
        ResultSet rs = null;
        List<String> primaryKeys = new ArrayList<>();
        try {
            rs = metaData.getPrimaryKeys(dbName, null, tableName);
            while (rs.next()) {
                primaryKeys.add(rs.getString(4));
            }
        } catch (Exception e) {
            log.error(e.getMessage());
        } finally {
            closeResult(rs);
        }
        return primaryKeys;
    }


    public  static List<ColumnMeta> getColumns(String jdbcUrl, String tableName, String username, String password) throws DataSourceException {
        Connection connection = null;
        try {
            connection = getConnection(jdbcUrl, username, password);
            if (null != connection) {
                DatabaseMetaData metaData = connection.getMetaData();
                return getColumns(tableName, metaData, jdbcUrl);
            }
        } catch (SQLException e) {
            throw new DataSourceException(e.getMessage() + ", jdbcUrl=" + jdbcUrl);
        } finally {
            releaseConnection(connection);
        }
        return null;
    }

    /**
     * 获取数据表列
     *
     * @param tableName
     * @param metaData
     * @return
     * @throws GlobalServiceException
     */
    private static List<ColumnMeta> getColumns(String tableName, DatabaseMetaData metaData, String jdbcUrl) throws GlobalServiceException {
        DatabaseTypeEnum databaseTypeEnum = DatabaseTypeEnum.urlOf(jdbcUrl);
        String dbName = getSchemaByJdbcUrl(jdbcUrl);
        ResultSet rs = null;
        List<ColumnMeta> columnList = new ArrayList<>();
        try {
            if (databaseTypeEnum == DatabaseTypeEnum.ORACLE) {
                dbName = null;
            }
            rs = metaData.getColumns(dbName, null, tableName, "%");
            while (rs.next()) {
                ColumnMeta columnDto = getColumnMeta(rs);
                columnList.add(columnDto);
            }
            List<String> primaryKeys = getPrimaryKeys(dbName, tableName, metaData);
            if(primaryKeys != null && primaryKeys.size() > 0){
                for(String priKey : primaryKeys){
                    for(ColumnMeta columnDto : columnList){
                        if(columnDto.getName().equalsIgnoreCase(priKey)){
                            columnDto.setPk(true);
                            break;
                        }else{
                            columnDto.setPk(true);
                        }
                    }
                }
            }
        } catch (Exception e) {
            throw new GlobalServiceException(ResultCodeEnum.DATA_SOURCE_EXCEPTION.getCode(), e.getMessage());
        } finally {
            closeResult(rs);
        }
        return columnList;
    }

    /**
     *  获取列的元数据，具体对应关系如下:
     *  1	：	库名
     * 3	：	表名
     * 4	：	列名
     * 5	：	对应java.sql.Types的值
     * 6	：	JDBC类型
     * 7	：	长度
     * 9	：	精度
     * 12	：	注释
     * 13	：	默认值
     * 17	：	顺序
     * 18	：	是否允许为空
     * @param rs
     * @return
     */
    private static ColumnMeta getColumnMeta(ResultSet rs) throws SQLException {
        ColumnMeta columnDto = new ColumnMeta();
        columnDto.setName(rs.getString(4));
        columnDto.setJdbcType(rs.getString(6));
        columnDto.setSize(rs.getInt(7));
        columnDto.setPrecision(rs.getInt(9));
        String isNullableStr = rs.getString(18);
        if(StringUtils.isNotEmpty(isNullableStr) && "NO".equalsIgnoreCase(isNullableStr)){
            columnDto.setNullable(false);
        }else{
            columnDto.setNullable(true);
        }
        columnDto.setComment(rs.getString(12));
        columnDto.setSqlTypesVal(rs.getInt(5));
        return columnDto;

    }


    /**
     * 获取数据源
     *
     * @param jdbcUrl
     * @param userName
     * @param password
     * @return
     * @throws DataSourceException
     */
    private static DataSource getDataSource(String jdbcUrl, String userName, String password) throws DataSourceException {
        jdbcUrl = setJdbcOfRemarkProperties(jdbcUrl);
        return  new SimpleDataSource(jdbcUrl, userName, password);
    }


    /**
     * 检查敏感操作
     *
     * @param sql
     * @throws GlobalServiceException
     */
    public static void checkSensitiveSql(String sql) throws GlobalServiceException {
        Pattern pattern = Pattern.compile(REG_SENSITIVE_SQL);
        Matcher matcher = pattern.matcher(sql.toLowerCase());
        if (matcher.find()) {
            String group = matcher.group();
            log.warn("Sensitive SQL operations are not allowed: {}", group.toUpperCase());
            throw new GlobalServiceException(ResultCodeEnum.DATA_SOURCE_EXCEPTION.getCode(), "Sensitive SQL operations are not allowed: " + group.toUpperCase());
        }
    }

    public static Connection getConnection(String jdbcUrl, String username, String password) throws DataSourceException {
        DataSource dataSource = getDataSource(jdbcUrl, username, password);
        Connection connection = null;
        try {
            connection = dataSource.getConnection();
        } catch (Exception e) {
            log.error("Get Connection Error,JdbcUrl:{}",jdbcUrl, e);
            connection = null;
        }
        try {
            if (null == connection || connection.isClosed() || !connection.isValid(5)) {
                log.info("connection is closed or invalid, retry get connection!");
                connection = dataSource.getConnection();
            }
        } catch (Exception e) {
            throw new DataSourceException("create connection error, jdbcUrl: " + jdbcUrl);
        }
        return connection;
    }

    private static void releaseConnection(Connection connection) {
        if (null != connection) {
            try {
                connection.close();
                connection = null;
            } catch (Exception e) {
                log.error("connection close error", e.getMessage());
            }
        }
    }


    public static void closeResult(ResultSet rs) {
        if (rs != null) {
            try {
                rs.close();
                rs = null;
            } catch (Exception e) {
                log.info(e.getMessage(), e);
            }
        }
    }

    public static boolean testConnection(String jdbcUrl, String username, String password) throws Exception {
        Connection connection = null;
        try {
            connection = getConnection(jdbcUrl, username, password);
            if (null != connection) {
                return true;
            }
            return false;
        } catch (Exception e) {
            throw e;
        } finally {
            releaseConnection(connection);
        }
    }

    public static JdbcTemplate jdbcTemplate(String jdbcUrl, String username, String password) throws DataSourceException {
        DataSource dataSource = getDataSource(jdbcUrl, username, password);
        JdbcTemplate jdbcTemplate = new JdbcTemplate(dataSource);
        jdbcTemplate.setFetchSize(1000);
        return jdbcTemplate;
    }

    public static  void executeBatch(String sql, Set<SimpleColumnMeta> headers, List<Map<String, Object>> datas
            , String jdbcUrl, String username, String password) throws DataSourceException {

        if (StringUtils.isEmpty(sql)) {
            log.info("execute batch sql is EMPTY");
            throw new DataSourceException("execute batch sql is EMPTY");
        }

        if (CollectionUtils.isEmpty(datas)) {
            log.info("execute batch data is EMPTY");
            throw new DataSourceException("execute batch data is EMPTY");
        }
        Connection connection = null;
        try {
            connection = getConnection(jdbcUrl, username, password);
            if (null != connection) {
                connection.setAutoCommit(false);
                try (PreparedStatement pstmt = connection.prepareStatement(sql)) {
                    //每1000条commit一次
                    int n = 10000;

                    for (Map<String, Object> map : datas) {
                        int i = 1;
                        for (SimpleColumnMeta queryColumn : headers) {
                            Object obj = map.get(queryColumn.getName());
                            switch (JdbcToJavaMappingEnum.toJavaType(queryColumn.getType())) {
                                case "Short":
                                    pstmt.setShort(i, null == obj ? (short) 0 : Short.parseShort(String.valueOf(obj).trim()));
                                    break;
                                case "Integer":
                                    pstmt.setInt(i, null == obj ? 0 : Integer.parseInt(String.valueOf(obj).trim()));
                                    break;
                                case "Long":
                                    pstmt.setLong(i, null == obj ? 0L : Long.parseLong(String.valueOf(obj).trim()));
                                    break;
                                case "BigDecimal":
                                    pstmt.setBigDecimal(i, (BigDecimal) obj);
                                    break;
                                case "Float":
                                    pstmt.setFloat(i, null == obj ? 0.0F : Float.parseFloat(String.valueOf(obj).trim()));
                                    break;
                                case "Double":
                                    pstmt.setDouble(i, null == obj ? 0.0D : Double.parseDouble(String.valueOf(obj).trim()));
                                    break;
                                case "String":
                                    pstmt.setString(i, (String) obj);
                                    break;
                                case "Boolean":
                                    pstmt.setBoolean(i, null == obj ? false : Boolean.parseBoolean(String.valueOf(obj).trim()));
                                    break;
                                case "Bytes":
                                    pstmt.setBytes(i, (byte[]) obj);
                                    break;
                                case "Date":
                                    java.util.Date date = (java.util.Date) obj;
                                    pstmt.setDate(i, DateJdbcUtils.toSqlDate(date));
                                    break;
                                case "DateTime":
                                    DateTime dateTime = (DateTime) obj;
                                    pstmt.setTimestamp(i, DateJdbcUtils.toTimestamp(dateTime));
                                    break;
                                case "Timestamp":
                                    pstmt.setTimestamp(i, (Timestamp) obj);
                                    break;
                                case "Blob":
                                    pstmt.setBlob(i, (Blob) obj);
                                    break;
                                case "Clob":
                                    pstmt.setClob(i, (Clob) obj);
                                    break;
                                default:
                                    pstmt.setObject(i, obj);
                            }
                            i++;
                        }

                        pstmt.addBatch();
                        if (i % n == 0) {
                            try {
                                pstmt.executeBatch();
                                connection.commit();
                            } catch (BatchUpdateException e) {
                            }
                        }
                    }

                    pstmt.executeBatch();
                    connection.commit();
                } catch (Exception e) {
                    connection.rollback(); //异常时进行回滚
                    //connection.setAutoCommit(true);
                    throw e;
                }
            }
        } catch (Exception e) {
            throw new GlobalServiceException(e.getMessage());
        } finally {
            releaseConnection(connection);
        }
    }

    public static String getKeywordPrefix(String jdbcUrl) {
        String keywordPrefix = "";
        CustomDataSourceConfig customDataSource = CustomDataSourceUtils.getInstance(jdbcUrl);
        if (null != customDataSource) {
            keywordPrefix = customDataSource.getKeyword_prefix();
        } else {
            DatabaseTypeEnum dataTypeEnum = DatabaseTypeEnum.urlOf(jdbcUrl);
            if (null != dataTypeEnum) {
                keywordPrefix = dataTypeEnum.getKeywordPrefix();
            }
        }
        return StringUtils.isEmpty(keywordPrefix) ? "" : keywordPrefix;
    }

    public static String getKeywordSuffix(String jdbcUrl) {
        String keywordSuffix = "";
        CustomDataSourceConfig customDataSource = CustomDataSourceUtils.getInstance(jdbcUrl);
        if (null != customDataSource) {
            keywordSuffix = customDataSource.getKeyword_suffix();
        } else {
            DatabaseTypeEnum dataTypeEnum = DatabaseTypeEnum.urlOf(jdbcUrl);
            if (null != dataTypeEnum) {
                keywordSuffix = dataTypeEnum.getKeywordSuffix();
            }
        }
        return StringUtils.isEmpty(keywordSuffix) ? "" : keywordSuffix;
    }

    public static String getAliasPrefix(String jdbcUrl) {
        String aliasPrefix = "";
        CustomDataSourceConfig customDataSource = CustomDataSourceUtils.getInstance(jdbcUrl);
        if (null != customDataSource) {
            aliasPrefix = customDataSource.getAlias_prefix();
        } else {
            DatabaseTypeEnum dataTypeEnum = DatabaseTypeEnum.urlOf(jdbcUrl);
            if (null != dataTypeEnum) {
                aliasPrefix = dataTypeEnum.getAliasPrefix();
            }
        }
        return StringUtils.isEmpty(aliasPrefix) ? "" : aliasPrefix;
    }

    public static String getAliasSuffix(String jdbcUrl) {
        String aliasSuffix = "";
        CustomDataSourceConfig customDataSource = CustomDataSourceUtils.getInstance(jdbcUrl);
        if (null != customDataSource) {
            aliasSuffix = customDataSource.getAlias_suffix();
        } else {
            DatabaseTypeEnum dataTypeEnum = DatabaseTypeEnum.urlOf(jdbcUrl);
            if (null != dataTypeEnum) {
                aliasSuffix = dataTypeEnum.getAliasSuffix();
            }
        }
        return StringUtils.isEmpty(aliasSuffix) ? "" : aliasSuffix;
    }


    /**
     * 过滤sql中的注释
     *
     * @param sql
     * @return
     */
    public static String filterAnnotate(String sql) {
        Pattern p = Pattern.compile(REG_SQL_ANNOTATE);
        sql = p.matcher(sql).replaceAll("$1");
        sql = sql.replaceAll(GlobalConstants.NEW_LINE_CHAR, GlobalConstants.SPACE).replaceAll("(;+\\s*)+", GlobalConstants.SEMICOLON);
        return sql;
    }

    public static String formatSqlType(String type) throws GlobalServiceException {
        if (!StringUtils.isEmpty(type.trim())) {
            type = type.trim().toUpperCase();
            String reg = "^.*\\s*\\(.*\\)$";
            Pattern pattern = Pattern.compile(reg);
            Matcher matcher = pattern.matcher(type);
            if (!matcher.find()) {
                return SqlTypeDefaultLengthEnum.getType(type);
            } else {
                return type;
            }
        }
        return null;
    }


    /**
     * 通过jdbc获取schema库名，
     * 如通过：  url: jdbc:mysql://127.0.0.1:3306/warehouse?allowMultiQueries=true
     *  获取名称：warehouse
     * @param jdbcUrl
     * @return
     */
    public static String getSchemaByJdbcUrl(String jdbcUrl){

        String schema = "";
        if(StringUtils.isNotEmpty(jdbcUrl)){
            if(jdbcUrl.indexOf("?") > 0){
                jdbcUrl = jdbcUrl.substring(0, jdbcUrl.indexOf("?"));
            }
            schema = jdbcUrl.substring(jdbcUrl.lastIndexOf("/") + 1, jdbcUrl.length());
            return schema;
        }else{
            log.warn("Get schema fail！ jdbcUrl is null!");
        }
        return schema;
    }

}
